#ifndef CONCURRENTPIPELINE_H
#define CONCURRENTPIPELINE_H

#include <QQueue>
#include <QMutex>
#include <QWaitCondition>
#include <QDebug>

template<typename T>
class ConcurrentPipeline {
private:
    QQueue<T> m_queue;
    mutable QMutex m_mutex;
    QWaitCondition m_condition;
    bool m_stopped = false;
    int m_maxSize = 10000;  // 防止内存溢出

public:
    ConcurrentPipeline() = default;

    // 生产者：放入数据
    bool push(const T &item) {
        QMutexLocker locker(&m_mutex);

        // 队列满时等待
        while (m_queue.size() >= m_maxSize && !m_stopped) {
            m_condition.wait(&m_mutex);
        }

        if (m_stopped) return false;

        m_queue.enqueue(item);
        m_condition.wakeOne();  // 唤醒一个消费者
        return true;
    }
    // 批量数据入队
    bool pushBatch(const QList<T> &items) {
        if (items.isEmpty()) return true;

        QMutexLocker locker(&m_mutex);
        // 检查是否有足够空间
        while ((m_queue.size() + items.size()) >= m_maxSize && !m_stopped) {
            qDebug() << "插入阻塞了。。。";
            m_condition.wait(&m_mutex);
        }
        if (m_stopped) return false;

        // 批量添加
        for (const T &item : items) {
            m_queue.enqueue(item);
        }

        m_condition.wakeAll();  // 唤醒所有等待的消费者
        return true;
    }

    // 消费者：取出数据（阻塞）
    T pop() {
        QMutexLocker locker(&m_mutex);

        // 队列空时等待
        while (m_queue.isEmpty() && !m_stopped) {
            m_condition.wait(&m_mutex);
        }

        if (m_stopped && m_queue.isEmpty()) {
            return T();  // 返回默认值
        }

        T item = m_queue.dequeue();
        m_condition.wakeOne();  // 唤醒可能等待的生产者
        return item;
    }
    // 批量数据出队（阻塞，直到有指定数量的数据或超时）
    QList<T> popBatch(int batchSize, int timeoutMs = -1) {
        QList<T> result;
        if (batchSize <= 0) return result;
        QMutexLocker locker(&m_mutex);

        // 等待直到有足够数据或超时
        if (timeoutMs >= 0) {
            // 带超时的等待
            QWaitCondition waitCondition;
            while (m_queue.size() < batchSize && !m_stopped) {
                if (!m_condition.wait(&m_mutex, timeoutMs)) {
                    return result; //返回空
                    break;
                }
            }
        } else {
            // 无限等待
            while (m_queue.size() < batchSize && !m_stopped) {
                qDebug() << "读取阻塞了。。。";
                m_condition.wait(&m_mutex);
            }
        }
        if (m_stopped && m_queue.isEmpty()) {
            return result;
        }

        // 取出数据（可能少于batchSize）
        int actualSize = qMin(batchSize, m_queue.size());
        for (int i = 0; i < actualSize; ++i) {
            result.append(m_queue.dequeue());
        }

        m_condition.wakeOne();
        return result;
    }

    // 消费者：尝试取出数据（非阻塞）
    bool tryPop(T &item) {
        QMutexLocker locker(&m_mutex);
        if (m_queue.isEmpty()) {
            return false;
        }
        item = m_queue.dequeue();
        m_condition.wakeOne();
        return true;
    }

    // 获取队列大小
    int size() const {
        QMutexLocker locker(&m_mutex);
        return m_queue.size();
    }

    bool isEmpty() const {
        QMutexLocker locker(&m_mutex);
        return m_queue.isEmpty();
    }

    // 停止管道
    void stop() {
        QMutexLocker locker(&m_mutex);
        m_stopped = true;
        m_condition.wakeAll();  // 唤醒所有等待的线程
    }

    bool isStopped() const {
        QMutexLocker locker(&m_mutex);
        return m_stopped;
    }

    void setMaxSize(int maxSize) {
        m_maxSize = maxSize;
    }
    int getMaxSize() const {
        return  m_maxSize;
    }
    //清空
    void clear() {
        QMutexLocker locker(&m_mutex);
        m_queue.clear();
        m_condition.wakeAll();
    }
};

#endif // CONCURRENTPIPELINE_H
